package beautiful.butterfly.drds.data_exchange.writer.util;


import beautiful.butterfly.drds.data_exchange.exception.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.util.*;
import beautiful.butterfly.drds.data_exchange.writer.Constant;
import beautiful.butterfly.drds.data_exchange.writer.Key;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public final class OriginalConfPretreatmentUtil
{


    public static void doPretreatment(Configuration originalConfig) throws SQLException
    {
        // 检查 username/password 配置（必填）
        originalConfig.getNecessaryValue(Key.username, DBUtilErrorCode.REQUIRED_VALUE);
        originalConfig.getNecessaryValue(Key.password, DBUtilErrorCode.REQUIRED_VALUE);

        doCheckBatchSize(originalConfig);

        simplifyConf(originalConfig);

        dealColumnConf(originalConfig);
        dealWriteMode(originalConfig);
    }

    public static void doCheckBatchSize(Configuration originalConfig)
    {
        // 检查batchSize 配置（选填，如果未填写，则设置为默认值）
        int batchSize = originalConfig.getInt(Key.batch_size, Constant.default_batch_size);
        if (batchSize < 1)
        {
            throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_VALUE, String.format(
                    "您的batchSize配置有误. 您所配置的写入数据库表的 batchSize:%s 不能小于1. 推荐配置范围为：[100-1000], 该值越大, 内存溢出可能性越大. 请检查您的配置并作出修改.",
                    batchSize));
        }

        originalConfig.set(Key.batch_size, batchSize);
    }

    public static void simplifyConf(Configuration configuration1)
    {
        List<Object> connections = configuration1.getList(Constant.connection, Object.class);

        int tableNum = 0;

        for (int i = 0, len = connections.size(); i < len; i++)
        {
            Configuration configuration = Configuration.from(connections.get(i).toString());

            String jdbcUrl = configuration.getString(Key.jdbc_url);
            if (StringUtils.isBlank(jdbcUrl))
            {
                throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.REQUIRED_VALUE, "您未配置的写入数据库表的 jdbcUrl.");
            }


            configuration1.set(String.format("%s[%d].%s", Constant.connection, i, Key.jdbc_url),
                    jdbcUrl);

            List<String> tables = configuration.getList(Key.table_name, String.class);

            if (null == tables || tables.isEmpty())
            {
                throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.REQUIRED_VALUE,
                        "您未配置写入数据库表的表名称. 根据配置DataX找不到您配置的表. 请检查您的配置并作出修改.");
            }

            // 对每一个connection 上配置的table 项进行解析
            List<String> expandedTables = TableExpandUtil.expand(tables);

            if (null == expandedTables || expandedTables.isEmpty())
            {
                throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.CONF_ERROR,
                        "您配置的写入数据库表名称错误. DataX找不到您配置的表，请检查您的配置并作出修改.");
            }

            tableNum += expandedTables.size();

            configuration1.set(String.format("%s[%d].%s", Constant.connection,
                    i, Key.table_name), expandedTables);
        }

        configuration1.set(Constant.table_number, tableNum);
    }

    public static void dealColumnConf(Configuration configuration, CreateConnectionFactory createConnectionFactory, String tableName) throws SQLException
    {
        List<String> userConfiguredColumnNameList = configuration.getList(Key.table_column_name_list, String.class);
        if (null == userConfiguredColumnNameList || userConfiguredColumnNameList.isEmpty())
        {
            throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_VALUE,
                    "您的配置文件中的列配置信息有误. 因为您未配置写入数据库表的列名称，DataX获取不到列信息. 请检查您的配置并作出修改.");
        }


        boolean getConnecttionWithoutRetry = configuration.getBoolean(Key.get_connecttion_without_retry, false);
        List<String> columnNameList;
        if (getConnecttionWithoutRetry)
        {
            columnNameList = Jdbcs1.getColumnNameList(createConnectionFactory.getConnecttionWithoutRetry(), tableName);
        } else
        {
            columnNameList = Jdbcs1.getColumnNameList(createConnectionFactory.getConnecttion(), tableName);
        }

        if (1 == userConfiguredColumnNameList.size() && "*".equals(userConfiguredColumnNameList.get(0)))
        {
            throw new IllegalArgumentException("*");
        } else if (userConfiguredColumnNameList.size() > columnNameList.size())
        {
            throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_VALUE,
                    String.format("您的配置文件中的列配置信息有误. 因为您所配置的写入数据库表的字段个数:%s 大于目的表的总字段总个数:%s. 请检查您的配置并作出修改.",
                            userConfiguredColumnNameList.size(), columnNameList.size()));
        } else
        {
            // 确保用户配置的 table_column_name_list 不重复
            ListUtil.makeSureNoValueDuplicate(userConfiguredColumnNameList);

            // 检查列是否都为数据库表中正确的列（通过执行一次 select table_column_name_list from tableName 进行判断）
            Jdbcs1.getColumnMetaDataList(createConnectionFactory.getConnecttion(), tableName, StringUtils.join(userConfiguredColumnNameList, ","));
        }
    }

    public static void dealColumnConf(Configuration configuration) throws SQLException
    {
        String jdbcUrl = configuration.getString(String.format("%s[0].%s",
                Constant.connection, Key.jdbc_url));

        String username = configuration.getString(Key.username);
        String password = configuration.getString(Key.password);
        String tableName = configuration.getString(String.format("%s[0].%s[0]", Constant.connection, Key.table_name));

        JdbcCreateConnectionFactory jdbcCreateConnectionFactory = new JdbcCreateConnectionFactory(jdbcUrl, username, password);
        dealColumnConf(configuration, jdbcCreateConnectionFactory, tableName);
    }

    public static void dealWriteMode(Configuration originalConfig)
    {
        List<String> columns = originalConfig.getList(Key.table_column_name_list, String.class);

        String jdbcUrl = originalConfig.getString(String.format("%s[0].%s",
                Constant.connection, Key.jdbc_url, String.class));

        // 默认为：insert 方式
        String writeMode = originalConfig.getString(Key.write_mode, "INSERT");

        List<String> valueHolders = new ArrayList<String>(columns.size());
        for (int i = 0; i < columns.size(); i++)
        {
            valueHolders.add("?");
        }

        boolean forceUseUpdate = false;


        String writeDataSqlTemplate = WriterUtil.getWriteTemplate(columns, valueHolders, writeMode, forceUseUpdate);

        log.info("Write data [\n{}\n], which jdbcUrl like:[{}]", writeDataSqlTemplate, jdbcUrl);

        originalConfig.set(Constant.insert_or_replace_template, writeDataSqlTemplate);
    }


}
